package com.example.springbootkafka.interceptor;

import com.example.springbootkafka.constants.BaseConstants;
import com.example.springbootkafka.service.ProducerService;
import io.netty.util.HashedWheelTimer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.SerializationUtils;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 生产者，延时消息拦截器
 *
 * @author huangchangjin
 * @date 2025/1/3 16:33
 */
@Slf4j
@Component
public class KafkaProducerInterceptorTtl implements ProducerInterceptor<String, String>, ApplicationContextAware {


    private static ApplicationContext applicationContext;

    // 秒
    private static final HashedWheelTimer time_wheel = new HashedWheelTimer(1000, TimeUnit.MICROSECONDS);

    // 缓存当前消息，防止重复发送
    private static final Map<String, ProducerRecord<String, String>> mq_cached = new ConcurrentHashMap<>();

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> p) {
        return selfHandleTtlMsg(p);
        //return p;
    }

    private ProducerRecord<String, String> selfHandleTtlMsg(ProducerRecord<String, String> p){
        long ttl = -1;
        String deadTopic = null;
        for (Header header : p.headers()) {
            if (header.key().equals(BaseConstants.TTL_KEY)) {
                //log.info("延时消息ttl: {}", ttl);
                ttl = (Long) SerializationUtils.deserialize(header.value());
            }
            if (header.key().equals(BaseConstants.DEAD_TOPIC_KEY)) {
                //log.info("延时消息topic: {}", BaseConstants.DEAD_TOPIC_KEY);
                deadTopic = (String) SerializationUtils.deserialize(header.value());
            }
        }

        if (deadTopic != null && ttl != -1) {
            //log.info("需要记录延时消息：{} 延时 {} 毫秒 到 topic= {}", LocalDateTime.now(), ttl, p.key());

            ProducerService producerService = applicationContext.getBean(ProducerService.class);

            // 1. 模拟 延时消息过期 ,直接不拦截
            ProducerRecord<String, String> cachedMq = mq_cached.get(p.key());
            if (cachedMq != null) {
//                log.info("重试的延时消息，无需重新拦截 延时 {} 毫秒 到 key = {}", ttl, p.key());
                return p;
            }


            // 2. 延时时间 10 秒钟内，可以用延时队列模拟延时消息
            boolean isOk = addMsgToDelayedQueue(p, ttl, producerService);
            if (!isOk) {
                // 立即 消费消息(把延时消息变为非延时消息)
                return new ProducerRecord<>(p.topic(), p.partition(), p.timestamp(), p.key(), p.value());
            }

            mq_cached.computeIfAbsent(p.key(), k -> p);
        }

        return p;
    }

    /**
     * 把消息推送到延时队列
     */
    private boolean addMsgToDelayedQueue(ProducerRecord<String, String> p, long ttl, ProducerService producerService) {
        boolean isOk = true;
        if (BaseConstants.MAX_TTL.compareTo(ttl) >= 0) {
            log.info("满足条件，入延时队列 {} 延时 {} 毫秒 到 key = {}", LocalDateTime.now(), ttl, p.key());
            try {
                time_wheel.newTimeout(task -> {
                    ProducerRecord<String, String> newR = new ProducerRecord<>(p.topic(), p.partition(), p.timestamp(), p.key(), p.value());
                    producerService.sendMessage(newR);
                }, ttl, TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                log.error("时间轮异常，{}", e.getMessage());
            }
            // 不再生产消息
        } else {
            log.error("不在延时时间支持内 ttl = {}, 按rabbitmq 的逻辑，应该直接消息", ttl);
            isOk = false;
        }
        return isOk;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        KafkaProducerInterceptorTtl.applicationContext = applicationContext;
    }
}
